package com.adcc.ags.cgw.managed.uplink;

import com.adcc.ags.cgw.configuration.GWFactory;
import com.adcc.ags.cgw.configuration.UserListFactory;
import com.adcc.ags.cgw.entity.mq.MQInfo;
import com.adcc.ags.cgw.entity.msg.MsgType;
import com.adcc.ags.cgw.entity.mq.MQState;
import com.adcc.ags.cgw.entity.user.GWMode;
import com.adcc.ags.cgw.entity.user.User;
import com.adcc.ags.cgw.managed.proxy.MsgProxy;
import com.adcc.ags.cgw.utils.Constant;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.MQConnectionPool;
import com.adcc.utility.mq.transfer.MQConnectionPoolFactory;
import com.adcc.utility.mq.transfer.MQTransfer;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.adcc.utility.mq.transfer.ibm.IBMMQTransfer;
import com.google.common.collect.Maps;
import io.dropwizard.lifecycle.Managed;
import io.dropwizard.setup.Environment;

import java.util.HashMap;
import java.util.Map;


/**
 * 网关上行处理Managed
 */
public class UplinkForward implements Managed {

    // 上行线程名称
    public static final String CONSTANT_UPLINK_THREAD_NAME = "Thread.Uplink";

    // 单例对象
    private static UplinkForward instance;

    // GWFactory
    private GWFactory gwFactory;

    // GW Host
    private String host;

    // GW Name
    private String name;

    // 网关运行模式(正常/应急)
    public GWMode mode = GWMode.NORMAL;

    // MQConfiguration
    private MQConfiguration configuration;

    // MQConnectionPool
    private MQConnectionPool pool;

    // MQTransfer
    private MQTransfer mqTransfer = new IBMMQTransfer();

    // MQ状态
    private MQInfo mqInfo = new MQInfo();

    // 上行处理线程
    private UplinkProcessRunnable upr;

    /**
     * 构造函数
     */
    public UplinkForward() {

    }

    /**
     * 单例方法
     * @return
     */
    public static UplinkForward getInstance() {
        if (instance == null) {
            instance = new UplinkForward();
        }
        return instance;
    }

    public String getHost() {
        return host;
    }

    public String getName() {
        return name;
    }

    public GWMode getMode(){
        return mode;
    }

    public void setMode(GWMode mode){
        this.mode = mode;
    }

    public MQInfo getMQInfo() {
        return mqInfo;
    }

    public int getMaxLength() {
        return gwFactory.getUplinkMsgMaxLength();
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment, GWFactory gwFactory) throws Exception {
        if (environment == null) {
            throw new NullPointerException("environment is null");
        }
        if (gwFactory == null) {
            throw new NullPointerException("gwFactory is null");
        }

        // 设置GW参数
        this.gwFactory = gwFactory;
        this.host = gwFactory.getHost();
        this.name = gwFactory.getName();

        // 设置MQ参数
        configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(gwFactory.getMqFactory().getHost(),
                gwFactory.getMqFactory().getPort(),
                gwFactory.getMqFactory().getQueueManager(),
                gwFactory.getMqFactory().getChannel());
        mqTransfer.setConfiguration(configuration);
        // 设置MQ连接池参数
        pool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
        ((IBMMQConnectionPool)pool).setActiveMode(gwFactory.getMqFactory().getActiveMode());
        ((IBMMQConnectionPool)pool).setTimeout(gwFactory.getMqFactory().getTimeout());
        ((IBMMQConnectionPool)pool).setMaxConnections(gwFactory.getMqFactory().getMaxConnections());
        ((IBMMQConnectionPool)pool).setMaxIdelConnections(gwFactory.getMqFactory().getMaxIdelConnections());
        mqTransfer.setConnectionPool(pool);
        pool.init(configuration);
        // 尝试连接MQ
        if (!connectMQ()) {
            mqInfo.setState(MQState.CONNECTING);
        }

        environment.lifecycle().manage(this);
    }

    /**
     * 连接MQ
     */
    public boolean connectMQ() {
        try {
            if (mqTransfer.isConnected()) {
                Log.info(UplinkForward.class.getName(), "gw connected host:" + gwFactory.getMqFactory().getHost() + " port:" + gwFactory.getMqFactory().getPort() + " channel:" + gwFactory.getMqFactory().getChannel() + " qm:" + gwFactory.getMqFactory().getQueueManager());
                mqInfo.setHost(gwFactory.getMqFactory().getHost());
                mqInfo.setPort(gwFactory.getMqFactory().getPort());
                mqInfo.setQm(gwFactory.getMqFactory().getQueueManager());
                mqInfo.setChannel(gwFactory.getMqFactory().getChannel());
                mqInfo.setState(MQState.CONNECTED);
                return true;
            }
            return false;
        } catch (Exception ex) {
            Log.error(UplinkForward.class.getName(), "connectMQ() error", ex);
            return false;
        }
    }

    public void sendMsgToBGS(String qName, byte[] msg, MsgType type, String sender) throws Exception{
        try {
            Map<String, Object> head = Maps.newHashMapWithExpectedSize(2);
            if (type == MsgType.Uplink) {
                type = MsgType.Ground;
            }
            head.put(Constant.STRING_MSG_TYPE, type.toString());
            head.put(Constant.STRING_MSG_SENDER, sender);
            Message message = new Message(head, msg);
            mqTransfer.sendQueue(qName, message);
        } catch (Exception ex) {
            Log.error(UplinkForward.class.getName(),"sendMsgToBGS() error",ex);
            throw ex;
        }
    }

    public void sendMsgToDB(byte[] msg) throws Exception {
        try {
            Message message = new Message(msg);
            mqTransfer.sendQueue(gwFactory.getDbForward(), message);
        } catch (Exception ex) {
            Log.error(UplinkForward.class.getName(), "sendMsgToDB() error", ex);
            throw ex;
        }
    }

    @Override
    public void start() {
        try {
            upr = new UplinkProcessRunnable();
            new Thread(upr, CONSTANT_UPLINK_THREAD_NAME).start();
        } catch (Exception ex) {
            Log.error(UplinkForward.class.getName(), "start() error", ex);
        }
    }

    @Override
    public void stop() throws Exception {
        try {
            if (upr != null) {
                upr.close();
            }
            pool.dispose();
            mqInfo.setState(MQState.DISCONNECTED);
        } catch (Exception ex) {
            Log.error(UplinkForward.class.getName(), "stop() error", ex);
        }
    }

    /**
     * 上行处理线程
     */
    class UplinkProcessRunnable implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close(){
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(UplinkProcessRunnable.class.getName(), "UplinkProcessRunnable start...");
                isStarted = true;
                while (isStarted) {

                    try {
                        // 如果存在用户列表则接收BGS发送的上行报文
                        if (UserListFactory.getInstance().getUserList().size() > 0) {
                            for (User user : UserListFactory.getInstance().getUserList()) {
                                // 接收BGS发送的上行报文
                                HashMap<String,Object> map = Maps.newHashMapWithExpectedSize(2);
                                map.put(Constant.STRING_RECEIVER_ACTOR_KEY1, mqTransfer);
                                map.put(Constant.STRING_RECEIVER_ACTOR_KEY2, user);
                                MsgProxy.getInstance().getMsgRecvActor().tell(map, MsgProxy.getInstance().getMsgRecvActor());
                            }
                        }
                        Thread.sleep(10);
                    } catch (Exception ex) {
                        Log.error(UplinkProcessRunnable.class.getName(), "process uplink message error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(UplinkProcessRunnable.class.getName(), "run() error",ex);
            }
        }
    }
}
